package com.qihoo.finance.backup.reader;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.qihoo.finance.backup.Result;
import com.qihoo.finance.backup.ResultUtils;
import com.qihoo.finance.backup.writer.BackWriter;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import javax.sql.DataSource;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * JDBC流式 reader
 */
public class JdbcStreamBackReader extends StreamBackReader {

    private static final Logger logger = LogManager.getLogger(JdbcStreamBackReader.class);
    private static final ObjectMapper objectMapper = new ObjectMapper();

    private DataSource dataSource;

    private String sql;

    private Integer rows;

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    public void setSql(String sql) {
        this.sql = sql;
    }

    public void setRows(Integer rows) {
        this.rows = rows;
    }

    @Override
    protected Result invoke() throws Exception {
        BackWriter backWriter = streamBacker.getBackWriter();
        Connection conn = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        try {
            backWriter.connect();
            conn = dataSource.getConnection();
            stmt = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            stmt.setFetchSize(Integer.MIN_VALUE);
            rs = stmt.executeQuery();
            List<String> rowList = new ArrayList<>();
            while (rs.next()) {
                //  检查备份器运行状态
                if (!streamBacker.isRunning()) {
                    return ResultUtils.success();
                }
                String nextJson = nextJson(rs);
                rowList.add(nextJson);
                if (rowList.size() >= rows) {
                    //  数据达到行数标准 执行写入
                    Result result = backWriter.write(rowList);
                    //  侦测写入过程是否存在错误
                    if (result.hasError()) {
                        return result;
                    }
                    rowList = new ArrayList<>();
                }
            }
            if (!rowList.isEmpty()) {
                Result result = backWriter.write(rowList);
                //  侦测写入过程是否存在错误
                if (result.hasError()) {
                    return result;
                }
            }
            return ResultUtils.success();
        } finally {
            close(conn, stmt, rs);
            if (backWriter != null) {
                backWriter.close();
            }
        }
    }

    private void close(Connection conn, PreparedStatement stmt, ResultSet rs) throws SQLException {
        if (conn != null) {
            conn.close();
        }
        if (stmt != null) {
            stmt.close();
        }
        if (rs != null) {
            rs.close();
        }
    }

    private String nextJson(ResultSet rs) throws SQLException, JsonProcessingException {
        ResultSetMetaData metaData = rs.getMetaData();
        int columnCount = metaData.getColumnCount();
        Map<String, Object> rowData = new HashMap<>(columnCount);
        for (int i = 1; i <= columnCount; i++) {
            rowData.put(metaData.getColumnName(i), rs.getObject(i));
        }
        return objectMapper.writeValueAsString(rowData);
    }
}
